#!/usr/bin/perl

package Migration;

use strict;
use warnings;
use lib "../libs";

use Exporter qw(import);
#our @EXPORT_OK = qw(test);

use JSON;
use File::Find;
use File::Basename;
use Const ();
use App ();
use Utils ();
use Utils qw(dd);
use Data::Dumper qw(Dumper);
use Log;

$DBI::Profile = 2; # 记录所有 SQL 执行时间

sub list
{
    my $migration_path = $App::CONF_PATH . $Const::DS . "migrations";
    my %migrations;
    find(sub {
        if (-f) {
            my $basename = basename($File::Find::name);
            $migrations{$basename} = decode_json(Utils::readFile($File::Find::name));
        }
    }, $migration_path);


    Log::info("Migration list:");
    for my $key (keys %migrations) {
        my %migration = %{ $migrations{$key} };
        print $migration{src}->{host}."\n";
        Log::info("> Src: type: $migration{src}->{host}, port: $migration{src}->{port}, user: $migration{src}->{user}, pass: $migration{src}->{pass}");
        Log::info("< Dst: type: $migration{dst}->{host}, port: $migration{dst}->{port}, user: $migration{dst}->{user}, pass: $migration{dst}->{pass}");

        unless (exists($migration{'tables'})) {
            Log::warn("- No tables to migrate!");
            next;
        }
        Log::info("- Tables:");
        my @tables = @{ $migration{'tables'} };
        for my $table (@tables) {
            print "\t$table\n";
        }
    }
}

sub run
{
    my $migration_path = $App::CONF_PATH. $Const::DS. "migrations";
    my %migrations;
    
    find(sub {
        if (-f) {
            my $basename = basename($File::Find::name);
            $migrations{$basename} = decode_json(Utils::readFile($File::Find::name));
        }
    }, $migration_path);

    for my $key (keys %migrations) {
        my %migration = %{ $migrations{$key} };
        Log::info("Migration: $key");

        my $max_processes = 3; # 最大并行进程数
        my @pids;
        my @tables = @{ $migration{'tables'} };
        my $table_count = scalar(@tables);

        for (my $i = 0; $i < $table_count; $i++) {
            my $table = $tables[$i];
            Log::info("==> Table: $table. in $i/$table_count");

            if (scalar(@pids) < $max_processes) {
                my $pid = fork();
                if ($pid == 0) {
                    my $mypid = $$;

                    Log::info("Child process [$mypid] started for table $table");
                    my $src_db = App::getDB($migration{'src'});
                    my $dst_db = App::getDB($migration{'dst'});

                    # migrate count
                    Log::info("====> Start migrate table [$table] data count.");
                    my $src_count = App::getCount($src_db, $table);
                    my $dst_count = App::getCount($dst_db, $table);
                    Log::info("Src count: $src_count, Dst count: $dst_count");
                    if ($src_count != $dst_count) {
                        Log::warn("Table [$table] data count not match!");
                        # exit 1;
                    }

                    # get columns
                    my $src_columns = $src_db->{dbh}->selectall_arrayref("SELECT column_name, data_type, data_length, nullable FROM user_tab_columns WHERE table_name = '${table}' ORDER BY column_id");

                    # migrate sum
                    Log::info("====> Start migrate table [$table] data sum.");
                    my @src_number_columns;
                    for my $column (@$src_columns) {
                        my ($column_name, $data_type, $data_length, $nullable) = @$column;
                        if ($data_type eq "NUMBER") {
                            push(@src_number_columns, $column_name);
                        }
                    }
                    Log::info("Src table [$table] number columns: ".join(", ", @src_number_columns));
                    for my $number_column (@src_number_columns) {
                        Log::info("======> Start migrate table [$table] column [$number_column] sum.");
                        my $src_sum = App::getSum($src_db, $table, $number_column);
                        my $dst_sum = App::getSum($dst_db, $table, $number_column);
                        Log::info("Src sum: $src_sum, Dst sum: $dst_sum");
                        if ($src_sum!= $dst_sum) {
                            Log::warn("Table [$table] column [$number_column] sum not match!");
                            # exit 1;
                        }
                    }

                    # migrate row
                    Log::info("-- Start migrate table [$table] data row.");
                    my @src_varchar_columns;
                    for my $column (@$src_columns) {
                        my ($column_name, $data_type, $data_length, $nullable) = @$column;
                        if ($data_type eq "VARCHAR2") {
                            push(@src_varchar_columns, $column_name);
                        }
                    }
                    Log::info("Src table [$table] varchar columns: ".join(", ", @src_varchar_columns));

                    my @conditions = ("id = 1");
                    my $base_row = App::getRow($src_db, $table, @conditions, 1);
                    dd($base_row);
                    for my $varchar_column (@src_varchar_columns) {
                        
                    }
                    
                    exit 0;
                } else {
                    push(@pids, $pid);
                }
            }
            
        }

        for my $pid (@pids) {
            waitpid($pid, 0);
        }
    }
}

1;

__END__
